package org.example.producer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 基于RocketMq分布式事务
 */
@Slf4j
public class TransactionProducer {
    @SneakyThrows
    public static void createTransactionProducer() {
        TransactionMQProducer mqProducer = new TransactionMQProducer("transaction_producer");
        mqProducer.setNamesrvAddr("192.168.31.104:9876");

        ThreadPoolExecutor ex = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS
                , new ArrayBlockingQueue<>(2000), (Runnable runnable) -> {
            Thread thread = new Thread(runnable);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        });
        // 本地事务执行线程池
        mqProducer.setExecutorService(ex);

        // 开启事务时发送prepare消息到MQServer，并客户端接收到ACK消息确认
        mqProducer.setTransactionListener(new TransactionListener() {
            // Message为prepare消息回执，根据prepare消息回执决定是否执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                log.info("本地事务执行。。。。");
                log.info("message is :{}", message);
                if (message.getTags().equals("1")) {
                    // 向MQServer提交二次确认，rollback则该perpare消息删除（不投放到消费队列）
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    log.info("事务执行成功，二次确认");
                    // 向MQServer提交二次确认，commit则该perpare消息投放到队列
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

            }

            // 当二次确认长时间未抵达MQ时，触发checkLocalTransaction
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                log.info("二次确认长时间未抵达MQ，messageExt：", messageExt);
                log.info("========查询对应本地事务的执行结果，进行事务回滚或提交逻辑======");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        mqProducer.start();

        for (int i = 0; i < 2; i++) {
            Message topicTransaction = new Message("TopicTransaction", String.valueOf(i), "=========".getBytes());
            // 发送perpare消息
            mqProducer.sendMessageInTransaction(topicTransaction, null);
            TimeUnit.MICROSECONDS.sleep(10);
        }

        for (int i = 0; i < 100; i++) {
            Thread.sleep(1000);
        }
        mqProducer.shutdown();
    }
}
